package com.demo.kettle.service.impl;


import cn.hutool.core.map.MapUtil;
import com.demo.kettle.req.KettleReq;
import com.demo.kettle.service.IKettleService;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.base.AbstractMeta;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleMissingPluginsException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.parameters.UnknownParamException;
import org.pentaho.di.core.variables.Variables;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 服务实现类
 *
 * @author yuanzheng
 * @since 2022-02-28
 */
@Slf4j
@Service
public class KettleServiceImpl implements IKettleService {

    @Override
    public String job(KettleReq kettleReq) {
        try {
            // 初始化执行环境
            KettleEnvironment.init();
            // 初始化Meta
            JobMeta jm = new JobMeta(kettleReq.getFilePath(), null);
            // 设置参数
            Map<String, Object> parameterMap = kettleReq.getParameterMap();
            setParameterValue(parameterMap, jm);
            // 初始化kjb
            final Job job = new Job(null, jm);
            AtomicBoolean jobStarted = new AtomicBoolean(true);
            int countOfCheckers = 100;
            HashMap<Integer, Future<StringBuilder>> checkers = new HashMap<>(16);
            ExecutorService executorService = Executors.newFixedThreadPool(countOfCheckers);
            for (int i = 0; i < countOfCheckers; i++) {
                Future<StringBuilder> future = executorService.submit(new Callable<StringBuilder>() {
                    private StringBuilder statusCollector = new StringBuilder();

                    @Override
                    public StringBuilder call() throws Exception {
                        while (jobStarted.get()) {
                            statusCollector.append(job.getStatus());
                            //add the sleep to reduce the count of logged status and prevent OOM
                            Thread.sleep(10);
                        }
                        return statusCollector;
                    }
                });
                checkers.put(i, future);
            }

            job.start();
            job.waitUntilFinished();
            jobStarted.set(false);

            for (Future<StringBuilder> checker : checkers.values()) {
                String checkerStr = checker.get().toString();
                System.out.println(checkOrderStatus(checkerStr));
            }

            // 关闭执行器
            executorService.shutdown();
            // 关闭执行环境
            KettleEnvironment.shutdown();
        } catch (ExecutionException e) {
            e.printStackTrace();
            log.error("ExecutionException {}", e.getMessage());
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.error("InterruptedException {}", e.getMessage());
        } catch (KettleXMLException e) {
            e.printStackTrace();
            log.error("KettleXMLException {}", e.getMessage());
        } catch (KettleException e) {
            e.printStackTrace();
            log.error("KettleException {}", e.getMessage());
        }
        return null;
    }

    @Override
    public String trans(KettleReq kettleReq) {
        try {
            // 初始化运行环境
            KettleEnvironment.init();
            // 初始化Meta
            TransMeta tm = new TransMeta(kettleReq.getFilePath(), new Variables());
            // 设置参数
            Map<String, Object> parameterMap = kettleReq.getParameterMap();
            setParameterValue(parameterMap, tm);
            // 初始化ktr
            Trans trans = new Trans(tm);
            trans.prepareExecution(new String[]{});
            trans.setPreview(true);
            trans.startThreads();
            trans.waitUntilFinished();

            // 关闭执行环境
            KettleEnvironment.shutdown();
        } catch (KettleMissingPluginsException e) {
            e.printStackTrace();
            log.error("KettleMissingPluginsException {}", e.getMessage());
        } catch (KettleXMLException e) {
            e.printStackTrace();
            log.error("KettleXMLException {}", e.getMessage());
        } catch (KettleException e) {
            e.printStackTrace();
            log.error("KettleException {}", e.getMessage());
        }
        return null;
    }

    private boolean checkOrderStatus(String status) {
        String[] tokens = {"Waiting", "Running", "Finished"};
        int offset = 0;
        for (String t : tokens) {
            while (status.startsWith(t, offset)) {
                offset += t.length();
            }
        }
        return offset == status.length();
    }

    private void setParameterValue(Map<String, Object> parameterMap, AbstractMeta meta) {
        if (MapUtil.isNotEmpty(parameterMap)) {
            parameterMap.forEach((k, v) -> {
                try {
                    meta.setParameterValue(k, v.toString());
                } catch (UnknownParamException e) {
                    log.error("UnknownParamException Key:{}.", k);
                }
            });
        }
    }
}
